import com.ververica.cdc.connectors.mysql.source.MySqlSource;
import com.ververica.cdc.connectors.mysql.table.StartupOptions;
import com.ververica.cdc.debezium.JsonDebeziumDeserializationSchema;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

import java.util.Properties;

public class flinkCDC {
    public static void main(String[] args) {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        Properties properties = new Properties();
        properties.setProperty("useSSL", "false");
        MySqlSource<String> mySqlSource = MySqlSource.<String>builder()
                .hostname("h102")
                .port(3306)
                .username("root")
                .password("j)k*x~y0)*n_L)!o{y?C`w(1rV_viw")
                .jdbcProperties(properties)
                .databaseList("edu_config")
                .tableList("edu_config.table_process")
                .startupOptions(StartupOptions.initial())
                .deserializer(new JsonDebeziumDeserializationSchema())
                .build();

        env
                .fromSource(mySqlSource, WatermarkStrategy.noWatermarks(), "cdc")
                .print(">>>");

        try {
            env.execute();
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
}
